package com.yujiahao.bigdata.test

import org.apache.spark.rdd.RDD
import org.apache.spark.{SparkConf, SparkContext}

object Spark01_TestHotTop10_1 {
  def main(args: Array[String]): Unit = {
    //TODO 1、获得Spark环境的连接

    val conf: SparkConf = new SparkConf().setAppName("TestTop10").setMaster("local[*]")
    val sc: SparkContext = new SparkContext(conf)

    //TODO 3、业务逻辑--需求第二种写法--优化操作，增加缓存，降低shuffle


    //TODO 需求一的问题：优化方案二
    // 1、RDD被重复使用--需要进行持久化操作。2、连接数据时，可能会导致迪卡尔乘积以及shuffle，性能会降低
    val rdd: RDD[String] = sc.textFile("data/user_visit_action.txt")
    //优化一缓存起来
    rdd.cache()
    //TODO 统计点击数量
    val clickCountData: RDD[(String, Int)] = rdd.filter(data => {
      val datas: Array[String] = data.split("_")
      datas(6) != "-1"
    }).map(
      data => {

        val datas: Array[String] = data.split("_")
        (datas(6), 1)
      }
    ).reduceByKey(_ + _)
    //TODO 统计下单数量
    val orderCountData: RDD[(String, Int)] = rdd.filter(
      data => {
        val datas: Array[String] = data.split("_")
        datas(8) != "null"
      }
    ).flatMap(
      data => {
        val datas: Array[String] = data.split("_")
        val ids: Array[String] = datas(8).split(",")
        ids.map((_, 1))
      }
    ).reduceByKey(_ + _)
    //TODO 统计支付数量
    val payCountData: RDD[(String, Int)] = rdd.filter(
      data => {
        val datas: Array[String] = data.split("_")
        datas(10) != "null"
      }
    ).flatMap(
      data => {
        val datas: Array[String] = data.split("_")
        //这里因为会有多个产品ID用逗号隔开，因此要将这一个字符串进行切割处理转换成字符串
        val ids: Array[String] = datas(10).split(",")
        ids.map((_, 1))
      }
    ).reduceByKey(_ + _)

    //TODO 优化二：连接数据时，可能会导致迪卡尔乘积以及shuffle，性能会降低，
    val unnRDD: RDD[(String, (Int, Int, Int))] = clickCountData.map {

      case (cid, clickrCnt) => {
        (cid, (clickrCnt, 0, 0))
      }
    }.union(
      orderCountData.map {

        case (cid, orderCnt) => {
          (cid, (0, orderCnt, 0))
        }
      }).union(
      payCountData.map {

        case (cid, payCnt) => {
          (cid, (0, 0, payCnt))
        }
      })
    //这里将结果进行两两聚合
    val value: RDD[(String, (Int, Int, Int))] = unnRDD.reduceByKey(
      (t1, t2) => {
        (t1._1 + t2._1, t1._2 + t2._2, t1._3 + t2._3)
      }
    )
    value.sortBy(_._2, false).take(10).foreach(println)
    //TODO 2、关闭Spark环境的连接
    sc.stop()

  }

}
